package com.xframe.event.bus.core.msg;


import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;

public class DirectReceiveMsg extends FanoutReceiveMsg {

    private String[] routingKeys;

    public DirectReceiveMsg(Connection msgConn, IMsgProc msgProc, String queueName, String exchange, String routingKeys) {
        super(msgConn, msgProc, queueName, exchange);
        this.routingKeys = routingKeys.split(",");
    }

    protected BuiltinExchangeType getBuiltinExchangeType(){
        return BuiltinExchangeType.DIRECT;
    }

    @Override
    public void init() {
        try{
            curChannel = msgConn.createChannel();
            curChannel.exchangeDeclare(this.exchange, getBuiltinExchangeType(), true);
            curChannel.queueDeclare(this.queueName, true, false, false, null);
            for(int i = 0; i < this.routingKeys.length; i++) {
                curChannel.queueBind(this.queueName, this.exchange, this.routingKeys[i]);
            }
            DefaultConsumer defaultConsumer = new DefaultReceiveConsumer(this);
            curChannel.basicConsume(queueName, this.autoAck, defaultConsumer);
        }catch (Exception exception){
            exception.printStackTrace();
        }
    }
}
